package own.test.service.impl;

import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.listener.MessageListener;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import own.test.service.KafkaService;

public class KafkaServiceImpl implements KafkaService {

	@Autowired
    private KafkaTemplate<String, String> kafkaTemplate;
	
	public void sendNonBlocking(final String data) {
	    //final ProducerRecord<String, String> record = createRecord(data);

	    ListenableFuture<SendResult<String, String>> future = kafkaTemplate.sendDefault(data);
	    future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {

	        @Override
	        public void onSuccess(SendResult<String, String> result) {
	            System.out.println("成功发送："+data);
	        }

	        @Override
	        public void onFailure(Throwable ex) {
	            System.out.println("发送失败："+data);
	            System.out.println("失败原因："+ex);
	        }

	    });
	}
	
	
	public void sendBlocking(final String data) {
	    //final ProducerRecord<String, String> record = createRecord(data);

	    try {
	    	kafkaTemplate.sendDefault(data).get(10, TimeUnit.SECONDS);
	    	System.out.println("成功发送："+data);
	    }
	    catch (ExecutionException e) {
	    	System.out.println("发送失败："+data);
            System.out.println("失败原因："+e);
	    }
	    catch (TimeoutException | InterruptedException e) {
	    	System.out.println("发送超时："+data);
	    }
	}

}
